// Copyright (c) 2023 Presto Labs Pte. Ltd.
// Author: ziyan

#include "coin2/exchange/bitmart/feed_v3/processor.h"

#include <string>
#include <utility>

#include "coin2/exchange/base/api_util/json_schema.h"
#include "coin2/exchange/bitmart/feed_v3/update.h"
#include "coin2/exchange/util/json_util.h"
#include "coin2/exchange/base/api_util/zlib_decompress.h"
namespace coin2::exchange::bitmart::feed_v3::impl {

void BitmartFeedProcessor::OnTopicRecord(
    const base::executor::TopicRecordPacket& packet) {
  upd_mgr_.onBeginPacket(packet);
  rapidjson::Document document;

  document.Parse(packet.payload.data());
  if (document.HasParseError()) {
    document.Parse(deflate_string(packet.payload).c_str());
  }

  if (document.HasParseError()) {
    THROW() << "JSON parse error: "
            << GetParseError_En(document.GetParseError())
            << packet.payload;
  }

  if (parser_config_.validate_schema() && schema_) {
    schema_->AcceptOrThrow(document);
  }

  std::string channel;
  std::string url;
  if (document.HasMember("table")) {
    channel = document["table"].GetString();
  } else if (document.HasMember("url")) {
    url = document["url"].GetString();
    std::vector<std::string> splited = Split(url, "?");
    CHECK_EQ(splited.size(), 2);
    if (StringEndsWith(splited[0], "kline")) {
      channel = "kline_rest";
    }
  }
  if (!CheckChannelFilter(channel) && !document.HasMember("data")) {
    upd_mgr_.onEndPacket(packet);
    return;
  }
  if (channel.compare(0, 10, "spot/depth") == 0) {
    for (const auto& book_json : document["data"].GetArray()) {
      std::string native_symbol = book_json["symbol"].GetString();
      auto book = upd_mgr_.GetBookMutable(native_symbol);
      if (book != nullptr) {
        BitmartFeedUpdateParser::ParseBook(book, book_json);
      }
      upd_mgr_.PublishBook();
    }
  } else if (channel == "spot/trade") {
    int last = document["data"].Size();
    int index = 1;
    for (const auto& trade_json : document["data"].GetArray()) {
      std::string nativa_symbol = trade_json["symbol"].GetString();
      if (observed_trade_symbols.count(nativa_symbol) == 0){
        // First trade message of symbol is outdated, should be dropped.
        observed_trade_symbols.insert(nativa_symbol);
        break;
      }
      Trade* trade = upd_mgr_.GetTradeMutable();
      trade->Clear();
      BitmartFeedUpdateParser::ParseTrade(trade_json, trade);
      upd_mgr_.PublishTrade(nativa_symbol, index != last);
      index++;
    }
  } else if (channel == "kline_rest") {
    std::map<std::string, std::string> parameters = base::feed::FeedParsingProcessor::GetRequestParameters(url);
    std::string native_symbol = parameters["symbol"];
    auto interval = BitmartFeedUpdateParser::ParseKlineInterval(parameters["step"]);
    auto* klines = upd_mgr_.GetKlineMutable(native_symbol, interval);
    if (klines != nullptr) {
      bool to_publish = false;
      // query is split up several requests
      int page = -1;
      if (parameters.count("page") > 0) {
        page = std::stod(parameters["page"]);
      }
      if ((page == 0) || (page == -1 && document["data"]["data"]["klines"].Size() > 10)) {
        // first long query
        klines->SnapshotClear();
      } else {
        klines->UpdateClear();
      }
      for (const auto& item : document["data"]["data"]["klines"].GetArray()) {
        bool freezed = BitmartFeedUpdateParser::ParseKlineRest(item, klines, interval, packet.timestamp);
        to_publish = to_publish || freezed;
      }
      if (to_publish) {
        upd_mgr_.PublishKline(native_symbol);
      }
    }
  } else {
    LOG(INFO) << "[Bitmart Feed Processor] unknown msg for " << packet.topic_name << ": "
              << packet.payload.data();
  }
  upd_mgr_.onEndPacket(packet);
}
}  // namespace coin2::exchange::bitmart::feed_v3::impl